package spark

import java.time.LocalDateTime

import org.apache.spark.SparkConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by zhangbn on 2018/09/14.
  */
object SocketListener {

  def main(args: Array[String]): Unit = {

    System.setProperty("hadoop.home.dir", "E:\\hadoop-common-2.7.3-bin-master")
    System.setProperty("HADOOP_USER_NAME", "admin")

    val sparkConf = new SparkConf()
      .setAppName("SocketTest")
      .setMaster("local[3]")

    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()
      .getOrCreate()

    spark.sqlContext.sql("use ods")
    spark.sqlContext.sql(
      """
        |create table IF NOT EXISTS ods.face_similarity(
        |passid_1 string,
        |passid_2 string,
        |similarity double
        |)STORED AS parquet TBLPROPERTIES('parquet.compression'='SNAPPY')
      """.stripMargin)

    //创建StreamingContext
    val ssc = new StreamingContext(spark.sparkContext, Seconds(5))
    val textStream = ssc.socketTextStream("10.82.27.100", 12345)
    println(s"[${LocalDateTime.now()}]  SparkStreaming-Socket监听程序启动.........................................")
    textStream.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        val rowRdd = rdd.flatMap(_.split("\\|")).map(_.split(",")).map(arr => Row(arr(0), arr(1), arr(2).toDouble))
        val schema = StructType(
          Seq(
            StructField("passid_1", StringType, true),
            StructField("passid_2", StringType, true),
            StructField("similarity", DoubleType, true)
          )
        )

        val rdd2DF = spark.createDataFrame(rowRdd, schema)

        rdd2DF.createOrReplaceTempView("temp_DF")

        val startTime = System.currentTimeMillis
        val count = rdd2DF.count()
        println(s"[${LocalDateTime.now()}]  COUNT=${count}........................")
        //spark.sqlContext.sql("select * from temp_DF").show(1)

        spark.sqlContext.sql(s"INSERT INTO TABLE ods.face_similarity select * from temp_DF")

        val endTime = System.currentTimeMillis
        println(s"[${LocalDateTime.now()}]  ${count}条数据保存OK........................耗时：${(endTime-startTime)/1000}s   速率：${count/(((endTime-startTime)/1000)+1)}条/秒")

      }
    })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
